package pl.edu.agh.inpema.core.queue.impl;

import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;

import pl.edu.agh.inpema.core.queue.Queue;
import pl.edu.agh.inpema.core.queue.QueueMessage;

public class QueueImpl implements Queue {

	private static final String user = ActiveMQConnection.DEFAULT_USER;
	private static final String password = ActiveMQConnection.DEFAULT_PASSWORD;
	private static final String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	private static boolean transacted = false;
	private static int ackMode = Session.AUTO_ACKNOWLEDGE;
	
	private Connection connection = null;
	private Session session = null;
	
	private Map<String, MessageConsumer> consumersMap = new HashMap<String, MessageConsumer>();
	private Map<String, MessageProducer> producersMap = new HashMap<String, MessageProducer>();
	
	private static Logger logger = Logger.getLogger(QueueImpl.class);
	
	public void startConnection() throws JMSException {
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
		connection = connectionFactory.createConnection();
		connection.start();
		session = connection.createSession(transacted, ackMode);
	}
	
	public void closeConnection() {
		try {
			for (MessageConsumer consumer : this.consumersMap.values())
				consumer.close();
			for (MessageProducer producer : this.producersMap.values())
				producer.close();
			session.close();
            connection.close();
        } catch (Throwable ignore) {
        }
	}
	
	public void createQueue(String analyzerName) throws JMSException {
		if (session == null)
			this.startConnection();
		Destination destination = session.createQueue("inpema."+analyzerName);
		MessageConsumer consumer = session.createConsumer(destination);
		MessageProducer producer = session.createProducer(destination);
		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		consumersMap.put(analyzerName, consumer);
		producersMap.put(analyzerName, producer);
	}
	
	public QueueMessage getMessageForAnalyzer(String analyzerName, int timeout) throws JMSException {
		MessageConsumer consumer = consumersMap.get(analyzerName);
		Message message = consumer.receive(timeout);
		if (message == null)
			return null;
		ObjectMessage objMsg = (ObjectMessage) message;
		//logger.info("get - " + (QueueMessage) objMsg.getObject());
		return (QueueMessage) objMsg.getObject();
	}
	
	public void sendMessageToAnalyzer(String analyzerName, QueueMessage queueMessage) throws JMSException {
		MessageProducer producer = producersMap.get(analyzerName);
		Message message = session.createObjectMessage(queueMessage);
		producer.send(message);
		//logger.info("put - " + queueMessage);
	}
}
